package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.Duration;

@Component
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder){
        //1. 订阅主题
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //2. 处理消息的格式
        stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);

                //key: 100 value: like:1
                return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
            }
        })
                .groupBy((key, value) -> key)  //3. 分组
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))  //4. 设置时间窗口
                .aggregate(new Initializer<String>() {
                    /**
                     * 用来做数据初始化的
                     *  每一个时间窗口都会执行一次且只执行一次
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {  //5. 自定义聚合函数

                    /**
                     * 完成真正的数据聚合
                     *    每一个消息来的时候都会执行一次
                     * @param key: 消息的key值 - 文章的id
                     * @param value - 消息的内容- LIKE:1 VIEW:1
                     * @param aggregate - 上一次允许的结果 如果是第一个消息 传递给你的就说初始化的结果
                     *                第一个消息: like:1  aggregate=COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0 -> COLLECTION:0,COMMENT:0,LIKES:1,VIEWS:0
                     *                第二个消息: view:1  aggregate=COLLECTION:0,COMMENT:0,LIKES:1,VIEWS:0 -> COLLECTION:0,COMMENT:0,LIKES:1,VIEWS:1
                     * @return
                     */

                    @Override
                    public String apply(String key, String value, String aggregate) {
                        //1. 判断value是否有数据 如果没有则直接返回上一次的结果
                        if(StringUtils.isEmpty(value)){
                            return aggregate;
                        }

                        //2. 把上一次的结果恢复为变量
                        int col = 0,com=0,lik=0,vie=0;
                        String[] strs = aggregate.split(","); // [COLLECTION:0, COMMENT:0,...]
                        for (String str : strs) {
                            String[] split = str.split(":"); //[COLLECTION,0]
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case VIEWS:
                                    vie = Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split[1]);
                                    break;
                                case COLLECTION:
                                    col = Integer.parseInt(split[1]);
                            }
                        }

                        //3. 解析value 然后value的类型来添加对应类型的值
                        String[] values = value.split(":"); //LIKE:1 -> [LIKE,1]
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])){
                            case VIEWS:
                                vie += Integer.parseInt(values[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(values[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(values[1]);
                                break;
                            case COLLECTION:
                                col += Integer.parseInt(values[1]);
                        }

                        //4. 把变量的数据转换成字符串格式的数据并返回
                        String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);

                        log.error("key=" + key + " format= " + format);

                        return format;
                    }
                }, Materialized.as("hot-atricle-stream-count-001")) //给聚合的结果取一个别名  select avg(score) as scores
                .toStream()  //6. 结果格式的转换
                .map((key, value) -> {
                    log.info("key:" + key + " : " +"value:" + value);
                    return new KeyValue<>(key.key(), formartObj(key.key(), value.toString())); //value还需要进一步的处理
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC); //7. 将结果发到新的topic中暂存

        return stream;
    }

    /**
     * 对统计的结果进行进一步的封装
     * @param key
     * @param value
     * @return
     */
    private String formartObj(String key, String value) { //value=COLLECTION:0,COMMENT:0,LIKES:1,VIEWS:1
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] strs = value.split(",");
        for (String str : strs) {
            String[] split = str.split(":");//[COLLECTION,0]

            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case COLLECTION:
                   mess.setCollect(Integer.parseInt(split[1]));
            }
        }

        return JSON.toJSONString(mess);
    }
}
